package com.macro.cloud.controller;


import org.apache.commons.collections4.queue.CircularFifoQueue;
import org.apache.commons.collections4.queue.SynchronizedQueue;
import org.springframework.core.task.TaskExecutor;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.ArrayList;
import java.util.List;
import java.util.Queue;
import java.util.Random;
import java.util.concurrent.*;
import java.util.stream.Collectors;

import static java.util.stream.Collectors.toList;

public class Demo2 {
    private volatile Boolean closed = false;

    /**
     * 体检报告统计线程池
     *
     * @return
     */

    public static TaskExecutor divExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(1);
        executor.setMaxPoolSize(100);
        executor.setQueueCapacity(100);
        executor.setKeepAliveSeconds(10000);
        executor.setThreadNamePrefix("div-xxx-");
        // CALLER_RUNS：不在新线程中执行任务线程,而是调用运行该任务的execute本身。此策略提供简单的反馈控制机制，能够减缓新任务的提交速度。
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.initialize();
        return executor;
    }

    /**
     * 使用定制的 Executor 配置 CompletableFuture
     *
     * @return
     */
    public List<String> findPrice4() throws InterruptedException {
        List<Shop> shops = getProds(30);
        //为“最优价格查询器”应用定制的执行器 Execotor
        ExecutorService executor = Executors.newFixedThreadPool(10,
                r -> {
                    Thread thread = new Thread(r);
                    //使用守护线程,使用这种方式不会阻止程序的关停
                    thread.setDaemon(true);
                    return thread;
                }
        );



     /*   CompletableFuture<Long> future = CompletableFuture.supplyAsync(() -> {
            doTaskWithMuchTime(null);
            System.out.println("run end ...");
            return System.currentTimeMillis();
        });*/

     /*   CompletableFuture<Void> future2 = CompletableFuture.supplyAsync(() -> {
            doTaskWithMuchTime(null);
            System.out.println("run end ...");
            return null;
        }, executor);
*/

        // TaskExecutor taskExecutor = divExecutor();
        //将执行器Execotor 作为第二个参数传递给 supplyAsync 工厂方法

        List<CompletableFuture<String>> futures = shops.stream()
                .map(shop -> CompletableFuture
                        .supplyAsync(() -> {
                            try {
                                int s = new Random().nextInt(15);
                                TimeUnit.SECONDS.sleep(s);

                                while (closed) {
                                    System.out.println();
                                    return "closed: " + shop.getName();
                                }
                                System.out.println("s:" + s);
                                TimeUnit.SECONDS.sleep(s);
                                if (shop.getName().equals("shop9")) {
                                    int i = 12 / 0;
                                }
                            } catch (Throwable ex) {
                                ex.printStackTrace();
                                //shutdownNow直接关闭活跃状态的所有线程，并返回阻塞队列中的任务集合。
                                //List<Runnable> runnables = executor.shutdownNow();
                                //System.err.println("runnables.size()  :" + runnables.size());
                                //executor.shutdown();
                                System.out.println("executor.isTerminated() " + executor.isTerminated());
                            /*    try {
                                    executor.awaitTermination(0,TimeUnit.SECONDS);
                                } catch (InterruptedException e) {
                                    e.printStackTrace();
                                }*/
                                return "error exist!";
                            }
                            //业务
                            String res = doTaskWithMuchTime(shop);
                            return res;
                        }, executor)
                        .whenComplete((result, ex) -> {
                            if (ex != null) {
                                System.err.println("exception!:" + " ----->threadnum:" + Thread.currentThread());
                            }
                            System.err.println("result:" + result + " -----> threadnum:" + Thread.currentThread());
                        })
                ).collect(toList());

        TimeUnit.SECONDS.sleep(4);
       /* closed = true;
        TimeUnit.SECONDS.sleep(4);
        //TimeUnit.SECONDS.sleep(1);
        closed = false;
        TimeUnit.SECONDS.sleep(4);
        closed = true;*/
        // executor.shutdown();
        //TimeUnit.SECONDS.sleep(20);
        System.out.println("executor.isTerminated() " + executor.isTerminated());
       /* List<String> list = futures.stream()
                .map(CompletableFuture::join)
                .collect(toList());*/

        List<CompletableFuture<String>> noticeList = futures.subList(0, 4);
        noticeList.stream().map(CompletableFuture::join).collect(Collectors.toList());
        //fifoQueue.remove(7);

        System.out.println("0-4 end!");
        TimeUnit.SECONDS.sleep(6);
        return null;
    }

    //随机生成数据
    public static List<Shop> getProds(int num) {
        List<Shop> shops = new ArrayList<>();
        for (int i = 0; i < num; i++) {
            double price = Math.random() * 20000;
            Shop shop = new Shop();
            shop.setName("shop" + i);
            shop.setPrice(price);
            shops.add(shop);
        }
        return shops;
    }

    //做一些比较浪费时间的事情
    public static String doTaskWithMuchTime(Shop shop) {
        // System.out.println("内部方法开始 :" + System.currentTimeMillis() + " 线程号:" + Thread.currentThread());

        long s = (long) new Random(1).nextInt(15) * 1000;
        try {
            Thread.sleep(s);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        //System.out.println("内部方法结束:" + System.currentTimeMillis() + " 线程号:" + Thread.currentThread());
        return "shopnameandprice: " + shop.getName() + " " + shop.getPrice();
    }




    public static void main(String[] args) throws InterruptedException {
        //private static volatile CircularFifoQueue<Integer> fifoQueue = new CircularFifoQueue<>(5);
        //fifo队列
        CircularFifoQueue<Integer> fifoQueue = new CircularFifoQueue<>(5);
        SynchronizedQueue syncFifoQueue = SynchronizedQueue.synchronizedQueue(fifoQueue);
        //同步的fifo队列
        syncFifoQueue.add(0);
        syncFifoQueue.add(2);
       /* for (int i = 0; i < 111; i++) {
            fifoQueue.add(i);
        }*/

        for (int i = 0; i < 111; i++) {
            syncFifoQueue.add(i);
        }


        boolean remove = fifoQueue.remove(7);
        boolean remove2 = fifoQueue.remove(109);
        boolean remove3 = fifoQueue.remove(11111);

        boolean remove11 = syncFifoQueue.remove(7);
        boolean remove22 = syncFifoQueue.remove(109);
        boolean remove33 = syncFifoQueue.remove(11111);

        System.out.println(1111);


        Demo2 demo2 = new Demo2();
        demo2.findPrice4();
    }


}


/**
 * 中断线程池的某个任务.
 */
class InterruptThread implements Runnable {
    private int num;

    public InterruptThread(int num) {
        this.num = num;
    }

    /*   public static void main(String[] args) throws InterruptedException {
           Thread interruptThread = new Thread(new InterruptThread(1));
           ScheduledFuture<?> t = ThreadPoolUtils.getInstance().getThreadPool().scheduleAtFixedRate(interruptThread, 0, 2, TimeUnit.SECONDS);
           InterruptThread interruptThread1 = new InterruptThread(2);
           ThreadPoolUtils.getInstance().getThreadPool().scheduleAtFixedRate(interruptThread1, 0, 2, TimeUnit.SECONDS);
           InterruptThread interruptThread2 = new InterruptThread(3);
           ThreadPoolUtils.getInstance().getThreadPool().scheduleAtFixedRate(interruptThread2, 0, 2, TimeUnit.SECONDS);
           Thread.sleep(5000);
           //终止正在运行的线程
            t.cancel(true);
           while (true) {
           }
       }
   */
    @Override
    public void run() {
        System.out.println("this is a thread" + num);
    }
}



